package flink_p1

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import scala.collection.JavaConverters.asScalaIteratorConverter

object FlinkTest_13_State_ListState {


  /**
   * 计算每一辆车经过的地点
   * 思路：使用
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream
      .filter(_.contains(' '))
      .map(data => {
        val zrr: Array[String] = data.split(" ")
        (zrr(0), zrr(1))
      }).keyBy(0)
      .map(new RichMapFunction[(String, String), (String, String)] {

        private var wzList: ListState[String] = _

        override def open(parameters: Configuration): Unit = {

          val desc = new ListStateDescriptor[String]("wzList", createTypeInformation[String])
          wzList = getRuntimeContext.getListState(desc)

        }

        override def map(in: (String, String)): (String, String) = {

          wzList.add(in._2)

          val builder = new StringBuilder()
          //iterator() 返回的是一个Java类型，需要asScala 转换为scala类型，同时需要导入隐私转换

          wzList.get().iterator().asScala.toList.foreach(builder.append(_))

          (in._1, builder.toString())

        }
      }).print()

    env.execute()
  }

}
